package com.gckj.dfds.backup;

import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.nio.file.StandardOpenOption;
import java.util.concurrent.TimeUnit;

/**
 * @program: filesystem
 * @description: 定时进行check_pointer
 * @author: Liu Weidong
 * @create: 2024-05-08 22:02
 **/
public class FSImageCheckPointer extends Thread{

    /**
     *  每隔两分钟生成一次 fs_image
     */
    public static final Integer CHECK_POINTER_SCHEDULE = 10 * 1000;


    private String lastFsImagePath = "";

    private FSNamesystem fsNamesystem;

    private BackUpNode backUpNode;

    private NameNodeClientRpc nameNodeClientRpc;

    public FSImageCheckPointer(BackUpNode backUpNode, FSNamesystem fsNamesystem, NameNodeClientRpc nameNodeClientRpc){
        this.fsNamesystem = fsNamesystem;
        this.backUpNode = backUpNode;
        this.nameNodeClientRpc = nameNodeClientRpc;
    }


    @Override
    public void run() {
        while (backUpNode.isRunning()) {
            try {
                if(!fsNamesystem.isFinish()) {
                    TimeUnit.MILLISECONDS.sleep(1000);
                    continue;
                }
                TimeUnit.MILLISECONDS.sleep(CHECK_POINTER_SCHEDULE);
                FsImage fsImage = this.fsNamesystem.getDirectory().getFsImage();
                if (fsImage.getTxid().equals(0L) || lastFsImagePath.lastIndexOf(fsImage.getTxid().toString()) > 0) {
                    System.out.println("本次check_pointer数据未发生变化，无需刷盘操作...");
                    continue;
                }
                if ("".equals(lastFsImagePath)) {
                   lastFsImagePath = "D://dfs//images//image-" + this.fsNamesystem.getCheckPointerTxid() + ".meta";
                }
                // 先删除之前生成的fsImage
                removeLastFsImage();
                // 再写入新的fsImage
                writeFsImageToDisk(fsImage);
                // 写入一次磁盘后则将fsImage上传到namenode
                uploadFsImage(fsImage);
                // 更新checkPointer的txid
                updateCheckPointerTxid(fsImage);
                System.out.println("一次检查结束。。。。。。。。。。。。。。。。。。。。。。");
                // 保存checkerPointerId信息用于重启后恢复数据
                saveCheckPointerInfo(fsImage);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    private void saveCheckPointerInfo(FsImage fsImage) {
        try {
            Long txid = fsImage.getTxid();
            long createTime = System.currentTimeMillis();
            Files.write(Paths.get("D://dfs//back_check_pointer_txid.meta"),
                    (txid + "_" + createTime).getBytes(StandardCharsets.UTF_8),
                    StandardOpenOption.CREATE, StandardOpenOption.WRITE);
        }catch (IOException e) {
            e.printStackTrace();
        }
    }

    private void updateCheckPointerTxid(FsImage fsImage) {
        nameNodeClientRpc.updateCheckPointerTxid(fsImage);
    }

    private void uploadFsImage(FsImage fsImage) {
        FsImageTransfer fsImageTransfer = new FsImageTransfer(fsImage);
        fsImageTransfer.start();
    }


    private void writeFsImageToDisk(FsImage fsImage){
        String destFilePath = "D://dfs//images//image-" + fsImage.getTxid() + ".meta";
        lastFsImagePath = destFilePath;
        FileOutputStream fos = null;
        FileChannel channel = null;
        try {
            fos = new FileOutputStream(destFilePath);
            channel = fos.getChannel();
            ByteBuffer byteBuffer = ByteBuffer.wrap(fsImage.getFsImageJson().getBytes(StandardCharsets.UTF_8)); // 创建一个1024字节大小的Buffer
            channel.write(byteBuffer);
            channel.force(true);
            System.out.println("fsImage文件成功写入磁盘。。。。。。。。。。。" + destFilePath);
        } catch (IOException e) {
           e.printStackTrace();
        } finally {
            if (fos != null) {
                try {
                    fos.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
            if(channel != null) {
                try {
                    channel.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    private void removeLastFsImage(){
        File file = new File(lastFsImagePath);
        if (file.exists()) {
            file.delete();
        }
    }
}
